package com.gitee.zhanyao.cloud.mq.consumer.rocketmq;

import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * mq消费者配置
 *
 * @author chendm
 * @create 2021-03-09 16:33
 */
@Slf4j
@ToString
@Configuration
public class MQConsumerConfigure {

    @Autowired
    private MQConsumeMsgListenerProcessor consumeMsgListenerProcessor;

    @Autowired
    private MQConsumerProperties mqConsumerProperties;

    /**
     * mq 消费者配置
     *
     * @return
     * @throws MQClientException
     */
    @Bean
    public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
        log.info("defaultConsumer 正在创建---------------------------------------");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqConsumerProperties.getGroupName());
        consumer.setNamesrvAddr(mqConsumerProperties.getNamesrvAddr());
        consumer.setConsumeThreadMin(mqConsumerProperties.getConsumeThreadMin());
        consumer.setConsumeThreadMax(mqConsumerProperties.getConsumeThreadMax());
        consumer.setConsumeMessageBatchMaxSize(mqConsumerProperties.getConsumeMessageBatchMaxSize());
        // 设置监听
        consumer.registerMessageListener(consumeMsgListenerProcessor);

        /**
         * 设置consumer第一次启动是从队列头部开始还是队列尾部开始
         * 如果不是第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        try {
            // 设置该消费者订阅的主题和tag，如果订阅该主题下的所有tag，则使用*
            consumer.subscribe("TEST_TOPIC", "TEST_TAG");
            consumer.start();
            log.info("consumer 创建成功 groupName={}, topic={},tag={},namesrvAddr={}", mqConsumerProperties.getGroupName(),
                    "TEST_TOPIC", "TEST_TAG", mqConsumerProperties.getNamesrvAddr());
        } catch (MQClientException e) {
            log.error("consumer 创建失败!");
        }
        return consumer;
    }
}
